package concurrent.masterandworker;

import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;
import lombok.experimental.Accessors;

import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Consumer;

/**
 * @description: 执行任务类
 * @author: kanggw
 * @date: 2023/7/3
 **/
@Data
@AllArgsConstructor
@Builder
@Accessors(chain = true)
public class Worker<T extends Task, R> {
    private LinkedBlockingQueue<T> taskQueue = new LinkedBlockingQueue<>();
    private int workerId;
    private static AtomicInteger index = new AtomicInteger(0);
    private Thread thread;

    public Worker() {
        this.workerId = index.getAndIncrement();
        thread = new Thread(this::run);
        thread.start();
    }

    private void run() {
        for (; ; ) {
            try {
                T task = this.taskQueue.take();
                task.setWorkerId(this.workerId);
                task.execute();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    public void submit(T task, Consumer<R> action) {
        try {
            //设置任务异步回调方法
            task.resultAction = action;
            taskQueue.put(task);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    }


}


